package com.lzq.mq.broker.core;

import com.lzq.mq.broker.cache.CommonCache;
import com.lzq.mq.broker.model.*;
import com.lzq.mq.broker.utils.FileNameUtil;
import com.lzq.mq.broker.utils.lock.PutMessageLock;
import com.lzq.mq.broker.utils.lock.UnfairReentrantLock;
import com.lzq.mq.common.constants.BrokerConstants;
import com.lzq.mq.common.dto.MessageDTO;
import lombok.Getter;
import lombok.Setter;

import java.io.File;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.nio.ByteBuffer;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.util.List;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MMap文件映射类, 提供创建文件MMap,通过MMap读写文件的操作
 */
public class CommitLogMMapFileModel extends AbstractMMapFileModel {

    /**
     * topic名称
     */
    @Getter
    @Setter
    private String topicName;
    /**
     * 添加消息的锁
     * 为什么不使用synchronized?
     * 1, 范围不够灵活
     * 2, synchronized回在用户态和内核态切换,性能不够理想
     */
    private PutMessageLock putMsgLock;

    /**
     * 映射主题为名为topicName的最新commitLog文件
     * @param topicName 主题名称
     */
    public void loadFileInMMap(String topicName, int startOffset, int mappedSize) throws IOException {

        this.topicName = topicName;

        int pos = CommonCache.getMqTopicModelMap().get(topicName).getCommitLogModel().getOffset().get();
        this.doMMap(getLatestCommitLogFile(), startOffset, mappedSize, pos);

        putMsgLock = new UnfairReentrantLock();
    }

    /**
     * 获取最新的commitLog文件绝对路径
     * @return commitLog文件绝对路径
     */
    private String getLatestCommitLogFile() {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topicName);
        if (mqTopicModel == null) {
            throw  new IllegalArgumentException("非法的主题:" + topicName);
        }
        // 判断当前的文件是否已经到达最大上限
        CommitLogModel commitLogModel = mqTopicModel.getCommitLogModel();
        long diff = commitLogModel.countDiff();

        if (diff == 0) {
            // 如果当前文件已经到达上限, 需要创建新的commitLogFile
            FilePath newCommitLogFile = createNewCommitLogFile(commitLogModel.getFileName());
            return newCommitLogFile.getFilePath();
        } else {
            return FileNameUtil.buildCommitLogFilePath(this.topicName, commitLogModel.getFileName());
        }
    }

    /**
     * 创建新的commitLog文件
     * @param oldCommitLogFileName 生成新文件前最新的文件名称
     * @return 新的commitLog文件名称
     */
    private FilePath createNewCommitLogFile(String oldCommitLogFileName) {

        String newFileName = FileNameUtil.incrFileName(oldCommitLogFileName);
        String newFilePath = FileNameUtil.buildCommitLogFilePath(topicName, newFileName);
        // 创建文件
        try {
            boolean newFile = new File(newFilePath).createNewFile();
        } catch (Exception e) {
            throw new RuntimeException("创建文件失败:" + newFilePath);
        }

        return new FilePath(newFileName, newFilePath);
    }

    /**
     * 写入commitLog消息 不刷盘
     * 等价于 writeContent(messageModel, false)
     */
    public void writeContent(MessageDTO message) throws IOException {
        writeContent(message, false);
    }

    /**
     * 写入commitLog消息
     * @param message 消息
     * @param force 是否强制刷盘
     * @throws IOException 抛出IO异常
     */
    public void writeContent(MessageDTO message, boolean force) throws IOException {

        // 定义一个对象, 管理当前各个topic的offset值: x, commitLog只做增量添加
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(this.topicName);
        if (null == mqTopicModel) {
            throw new IllegalArgumentException("非法的topicName:" + this.topicName);
        }
        CommitLogModel commitLogModel = mqTopicModel.getCommitLogModel();
        if (null == commitLogModel) {
            throw new IllegalArgumentException("非法的commitLogModel");
        }
        // 高并发写入, 需要考虑多线程的情况, 需要保证线程安全(可能出现消息覆盖的问题)
        // offset 可以通过AtomicLog去管理, 但是无法保证消息的顺序性, 故必须加锁 ! 加什么锁！！！
        // todo 是否需要分布式锁?
        putMsgLock.lock();
        // 定位到当前commitLog文件最新位置, 记录当前文件是否写满, 如果写满创建信的文件, 并做新的mmap映射
        // 如果当前文件没有写满, 对content内容做一层封装, 判断是否会草纸当前commitLog文件是否写满。如果不会, 则选择当前commitLog, 否则创建新的commitLog文件, 并做新的mmap映射
        // 注意: 需要放在锁的范围以内
        CommitLogElementModel messageModel = new CommitLogElementModel(message.getBody());
        this.checkCommitLogHashEnableSpace(messageModel);           // 保证commitLog能够装下消息
        byte[] contentBytes = messageModel.getContent();
        writeByteBuffer.put(contentBytes);                         // 添加消息

        AtomicInteger currCLOffset = commitLogModel.getOffset();
        int msgLength = contentBytes.length;
        if (force) {
            writeByteBuffer.force();
        }
        /*
        将消息分配到消费队列ConsumeQueue中
         */
        this.dispatcher(message, currCLOffset.get());                   // 分配消息(偏移量应该是起始位置)
        currCLOffset.addAndGet(msgLength);                          // 调整commitLog消息偏移量
        putMsgLock.unlock();
    }

    /**
     * 读取commitLog文件
     * @param readOffset 读取的起始位置
     * @param size 读取的尺寸
     * @return 读取的内容
     */
    public byte[] readContent(int readOffset, int size) {
        return read(readOffset, size);
    }

    /**
     * 将消息的引用添加到消费队列中
     * @param msgDTO 消息
     * @param msgOffset 消息的起始位置
     */
    private void dispatcher(MessageDTO msgDTO, int msgOffset) {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(topicName);
        if (null == mqTopicModel) {
            throw new IllegalArgumentException("非法的topicName:" + this.topicName);
        }
        ConsumeQueueElementModel consumerQueueElementModel = new ConsumeQueueElementModel();
        consumerQueueElementModel.setCommitLogFileIndex(Integer.parseInt(mqTopicModel.getCommitLogModel().getFileName()));
        consumerQueueElementModel.setMsgOffset(msgOffset);
        consumerQueueElementModel.setMsgLength(msgDTO.getBody().length);
        // 将消息引用添加到消费队列
        List<ConsumeQueueMMapFileModel> consumeQueueMMapFileModelList = CommonCache.getConsumeQueueMMapFileModelManager()
                        .getTopicConsumeQueueMMapFileModelList(topicName);                          // 获取当前topic的队列
        // 在队列列表中选择一个队列装入消息引用
        int queueId;
        if(msgDTO.getQueueId() != null && msgDTO.getQueueId() >=0) {
            queueId = msgDTO.getQueueId();
        } else {
            // 可以在这里自由扩展不同的消息分派策略
            int queueSize = mqTopicModel.getQueueList().size();
            queueId = new Random().nextInt(queueSize);
        }
        ConsumeQueueMMapFileModel consumeQueueMMapFileModel = consumeQueueMMapFileModelList.stream().filter(
                queueModel -> queueModel.getQueueId().equals(queueId)
        ).findFirst().orElse(null); // 选择消费队列
        if (null == consumeQueueMMapFileModel) {
            throw new IllegalArgumentException("非法的队列ID:" + queueId);
        }
        // 写入引用
        byte[] cqContent = consumerQueueElementModel.coverToBytes();
        consumeQueueMMapFileModel.writeContent(cqContent);
        // 刷新偏移量
        ConsumeQueueModel consumeQueueModel = mqTopicModel.getQueueList().get(queueId);
        consumeQueueModel.getLatestOffset().addAndGet(cqContent.length);
    }


    /**
     * 判断当前的CommitLo最新文件分片是否还能写文件, 如果不能, 创建新的commitLogFile
     * @param commitLogElementModel 需要存储的消息
     */
    private void checkCommitLogHashEnableSpace(CommitLogElementModel commitLogElementModel) throws IOException {
        MqTopicModel mqTopicModel = CommonCache.getMqTopicModelMap().get(this.topicName);
        CommitLogModel commitLogModel = mqTopicModel.getCommitLogModel();
        long diff = commitLogModel.countDiff();
        if (diff < commitLogElementModel.getSize()) {
            // 剩余空间不足, 需要创建新的commitLogFile
            // 这样可能造成空间浪费！
            FilePath newCommitLogFile = createNewCommitLogFile(commitLogModel.getFileName());
            // 映射新的文件路径
            this.doMMap(newCommitLogFile.getFilePath(), 0, BrokerConstants.COMMITLOG_DEFAULT_MMAP_SIZE, 0);
            // 因当前topic有新的commitLog创建, 故需要刷新topic的最新commitLog信息
            commitLogModel.setFileName(newCommitLogFile.getFileName());
            commitLogModel.setOffset(new AtomicInteger(0));
            commitLogModel.setOffsetLimit(Long.valueOf(BrokerConstants.COMMITLOG_DEFAULT_MMAP_SIZE));
        }
    }

    /**
     * 清理内存映射
     */
    public void clean() {
        if (writeByteBuffer == null || !writeByteBuffer.isDirect() || writeByteBuffer.capacity() == 0) {
            return;
        }

        // JDK 9及以后，直接调用sun.misc.Unsafe的invokeCleaner方法更稳定
        try {
            Class<?> unsafeClass = Class.forName("sun.misc.Unsafe");
            Method cleanerMethod = unsafeClass.getMethod("invokeCleaner", ByteBuffer.class);
            Object unsafe = getUnsafeInstance(unsafeClass);
            cleanerMethod.invoke(unsafe, writeByteBuffer);
            System.out.println("通过sun.misc.Unsafe实现clean");
        } catch (Exception e) {
            // Fallback 回退到原有的反射调用方式，兼容老版本JDK
            // 获取attachment并调用cleaner的clean方法
            invoke(invoke(viewed(writeByteBuffer), "cleaner"), "clean");
        }
    }

    /**
     * 获取Unsafe的实例（通过反射方式，避免直接依赖）
     */
    private Object getUnsafeInstance(Class<?> unsafeClass) throws Exception {
        try {
            // JDK 9+获取Unsafe实例的新方法
            Method getUnsafe = unsafeClass.getDeclaredMethod("getUnsafe");
            getUnsafe.setAccessible(true);
            return getUnsafe.invoke(null);
        } catch (Exception e) {
            // JDK 8及以下回退，使用反射获取theUnsafe字段
            Field f = unsafeClass.getDeclaredField("theUnsafe");
            f.setAccessible(true);
            return f.get(null);
        }
    }

    /**
     * 反射调用方法
     * @param target 目标对象
     * @param methodName 方法名
     * @param args 参数
     * @return 方法返回值
     */
    private Object invoke(final Object target, final String methodName, final Class<?>... args) {
        return AccessController.doPrivileged((PrivilegedAction<Object>) () -> {
            try {
                Method method = method(target, methodName, args);
                method.setAccessible(true);
                return method.invoke(target);
            } catch (Exception e) {
                throw new IllegalStateException(e);
            }
        });
    }

    /**
     * 反射获取target对象中的methodName方法
     * @param target 目标对象
     * @param methodName 方法名
     * @param args 参数
     * @return  方法
     * @throws NoSuchMethodException 抛出方法不存在异常
     */
    private Method method(Object target, String methodName, Class<?>[] args)
            throws NoSuchMethodException {
        try {
            return target.getClass().getMethod(methodName, args);
        } catch (NoSuchMethodException e) {
            return target.getClass().getDeclaredMethod(methodName, args);
        }
    }

    /**
     * 这个方法用于找到DirectByteBuffer中的att属性, 这个属性即对应真实的用于存储内容的内存
     * @param buffer DirectByteBuffer
     * @return 真正存储内容的内存
     */
    private ByteBuffer viewed(ByteBuffer buffer) {
        String methodName = "viewedBuffer";
        Method[] methods = buffer.getClass().getMethods();
        for (Method method : methods) {
            if (method.getName().equals("attachment")) {
                methodName = "attachment";
                break;
            }
        }

        ByteBuffer viewedBuffer = (ByteBuffer) invoke(buffer, methodName);
        if (viewedBuffer == null)
            return buffer;
        else
            return viewed(viewedBuffer);
    }
}
